package com.bff.gaia.mix.examples.re;

import com.alibaba.fastjson.JSONObject;
import com.bff.gaia.api.common.functions.MapFunction;
import com.bff.gaia.api.common.functions.ReduceFunction;
import com.bff.gaia.api.common.functions.RichFlatMapFunction;
import com.bff.gaia.api.common.functions.RichMapFunction;
import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.TimeCharacteristic;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.xjoin.client.hbase.core.table.HbaseSideTableInfo;
import com.bff.gaia.mix.api.xjoin.client.hbase.parse.HbaseSideTableInfoBuilder;

import com.bff.gaia.mix.api.xjoin.core.enums.ECacheType;
import com.bff.gaia.mix.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.types.Row;
import com.bff.gaia.util.Collector;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class HBaseExample {
	public static void main(String[] args) throws Exception {
		MixStreamExecutionEnvironment env = MixStreamExecutionEnvironment.getExecutionEnvironment();
		env.setMixStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

		final String hbaseHost = "localhost";

		HbaseSideTableInfo item = HbaseSideTableInfoBuilder
			.buildHbaseSideTableInfo()
			.setHost(hbaseHost)
			.setParent("cf")
			.setTableName("item")
			.addfieldInfos(new String[]{
				"item_id string",
				"item_expo_id string",
				"item_category string",
				"item_category_level1 string",
				"item_seller_city string",
				"item_seller_prov string",
				"item_purch_level string",
				"item_gender string",
				"item_buyer_age string",
				"item_style_id string",
				"item_material_id string",
				"item_pay_class string",
				"item_brand_id string",
				"item__i_shop_id_ctr string",
				"item__i_brand_id_ctr string",
				"item__i_category_ctr string",
			})
			.setPrimaryKeys("item_id")
			.enableQualifierQuery()
			.finish();

		HbaseSideTableInfo click = HbaseSideTableInfoBuilder
			.buildHbaseSideTableInfo()
			.setHost(hbaseHost)
			.setParent("cf")
			.setTableName("click")
			.addfieldInfos(new String[]{
				"user_id string",
				"item_id string",
				"flag string",
				"date string"
			})
			.enableQualifierQuery()
			.finish();

		DataMixStream<String> userIDStream = env.fromElements("1", "3", "5");

		DataMixStream<Row> userIDRowStream = userIDStream.map(new MapFunction<String, Row>() {
			@Override
			public Row map(String value) throws Exception {
				Row row = new Row(1);
				row.setField(0, value);
				return row;
			}
		}).returns(new RowTypeInfo(TypeInformation.of(String.class)));

		DataMixStream<Row> clickStream = userIDRowStream
			.xjoin(click)
			.projectLeftKey(0)
			.projectRightKey(0)
			.projectRightField(0, 1)
			.setCacheType(ECacheType.ALL)
			.apply(EJoinType.EQUAL)
			.setParallelism(1);

		DataMixStream<Row> i2iStream = clickStream
			.flatMap(new i2iRichMapFunction())
			.returns(new RowTypeInfo(
				TypeInformation.of(String.class),
				TypeInformation.of(String.class),
				TypeInformation.of(Integer.class)));

//		i2iStream.keyBy(0).keyBy(1).reduce(new ReduceFunction<Row>() {
//			@Override
//			public Row reduce(Row value1, Row value2) throws Exception {
//				Row row = new Row(3);
//				row.setField(0, value1.getField(0));
//				row.setField(1, value1.getField(1));
//				row.setField(2, (Integer) value1.getField(2) + (Integer) value2.getField(2));
//				return row;
//			}
//		}).print();


		DataMixStream itemStream = i2iStream
			.xjoin(item)
			.projectLeftKey(2)
			.projectLeftField(0, 2, 3)
			.projectRightKey(0)
			.setCacheType(ECacheType.LRU)
			.apply(EJoinType.EQUAL)
			.setParallelism(1);
		itemStream.print();

		env.execute();
	}

	public static class i2iRichMapFunction extends RichFlatMapFunction<Row, Row> {

		private Map<String, Map<String, Object>> i2i;

		public List<Cell> getDataByRowKey(String tableName, String rowKey) throws IOException {
			Connection connection = HBaseUtil.getHbaseConnection();
			Get get = new Get(rowKey.getBytes());
			HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
			Result result = table.get(get);
			List<Cell> cells = result.listCells();
			connection.close();
			return cells;
		}

		public Map<String, Map<String, Object>> getI2i() throws IOException {
			Map<String, Map<String, Object>> i2i = new HashMap<>();
			long time = System.currentTimeMillis();
			SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
			String date = df.format(time);
			try {

				List<Cell> cells = getDataByRowKey("i2i", date);
				String value = "";

				for (Cell cell : cells) {
					value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()) + "\t" + value;
				}
				Map<String, Object> i2iJSONMap = JSONObject.parseObject(value);
				for (String key : i2iJSONMap.keySet()) {
					Map<String, Object> itemMap = JSONObject.parseObject(i2iJSONMap.get(key).toString());
					i2i.put(key, itemMap);
				}

			} catch (Exception e) {
				System.out.println("table i2i for date " + date + "is not exist, please update!");
			}
			return i2i;
		}

		@Override
		public void open(Configuration parameters) throws Exception {
			i2i = getI2i();
		}

		@Override
		public void flatMap(Row value, Collector<Row> out) throws Exception {
			String item_id = (String) value.getField(1);
			if (i2i.containsKey(item_id)) {
				Map<String, Object> i2iItem = i2i.get(item_id);
				for (String key : i2iItem.keySet()) {
					Row row = new Row(3);
					row.setField(0, value.getField(0));
//					row.setField(1, value.getField(1));
					row.setField(1, key);
					row.setField(2, i2iItem.get(key));
					out.collect(row);
				}
			} else {
				Row row = new Row(3);
				row.setField(0, value.getField(0));
//				row.setField(1, value.getField(1));
				row.setField(1, value.getField(1));
				row.setField(2, 3);
				out.collect(row);
			}
		}
	}
}